package com.up.kafka.serializable;

import com.up.kafka.utils.HeaderKey;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.UnsupportedEncodingException;
import java.util.Map;

/**
 * @author: baiMl
 * @title: MultiTypeDeserializer
 * @projectName: 海联技术中台V2.0
 * @depart 软件事业部
 * @description:
 * @date: 2022/5/18 15:29
 */
public class MultiTypeDeserializer implements Deserializer<Object> {
    private String encoding = "UTF8";

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        String propertyName = isKey ? "key.deserializer.encoding" : "value.deserializer.encoding";
        Object encodingValue = configs.get(propertyName);
        if (encodingValue == null) {
            encodingValue = configs.get("deserializer.encoding");
        }

        if (encodingValue instanceof String) {
            encoding = (String) encodingValue;
        }

    }

    @Override
    public Object deserialize(String topic, Headers headers, byte[] data) {
        if (headers == null) {
            return deserialize(topic, data);
        }
        Iterable<Header> customHeader = headers.headers(HeaderKey.HEADER_DATA_PACKET_TYPE);
        if (customHeader == null) {
            return deserialize(topic, data);
        }
        for (Header header : customHeader) {
            byte[] value = header.value();
            if (value == null || value.length <= 0) {
                break;
            }
            switch (value[0]) {
                case HeaderKey.BYTE_ARRAY_VALUE:
                    return data;
                case HeaderKey.STRING_VALUE:
                    return deserialize(topic, data);
                default:
                    break;
            }
            break;
        }
        return deserialize(topic, data);
    }

    @Override
    public Object deserialize(String topic, byte[] data) {
        try {
            if (data == null) {
                return null;
            } else {
                return new String(data, encoding);
            }
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + encoding);
        }
    }
}
